package com.zhss.dfs.backupnode.server;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * fsimage文件checkpoint组件
 */
public class FSImageCheckPointer extends Thread {
    /**
     * checkpoint 操作时间间隔
     */
    public static final Integer CHECKPOINT_INTERVAL = 30 * 1000;

    private BackupNode backupNode;

    private FSNameSystem nameSystem;

    private String lastFSImageFile;
    private NameNodeRpcClient namenode;

    public FSImageCheckPointer(BackupNode backupNode, FSNameSystem nameSystem, NameNodeRpcClient namenode) {
        this.backupNode = backupNode;
        this.namenode = namenode;
        this.nameSystem = nameSystem;
    }

    @Override
    public void run() {
        System.out.println("fsimage checkpoint定时调度线程启动....");
        while (backupNode.isRunning()) {
            try {
                Thread.sleep(CHECKPOINT_INTERVAL);
                System.out.println("准备执行checkpoint操作，写入fsimage文件......");
                //触发checkpoint
                doCheckPoint();

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void uploadFSImageFile(FSImage fsimage) {
        FSImageUploader fsImageUploader = new FSImageUploader(fsimage);
        fsImageUploader.start();
    }

    private void doCheckPoint() throws IOException {
        FSImage fsimage = nameSystem.getFSImage();

        this.removeLastFsimageFile();
        writeFsImageFile(fsimage);
        uploadFSImageFile(fsimage);
        updateCheckpointTxid(fsimage);
        return;
    }

    /**
     * 更新checkpoint txid
     *
     * @param fsimage
     */
    private void updateCheckpointTxid(FSImage fsimage) {
        namenode.updateCheckpointTxid(fsimage.getMaxTxid());
    }

    private void writeFsImageFile(FSImage fsimage) throws IOException {

        ByteBuffer byteBuffer = ByteBuffer.wrap(fsimage.getFsimageJson().getBytes());
        FileChannel channel = null;
        FileOutputStream fileOutputStream = null;
        RandomAccessFile fsimageFile = null;
        File path = new File("/Users/jinsimin/dfs/backupnode");
        if (!path.exists()) {
            path.createNewFile();
        }
        String filename = "fsimagef-" + fsimage.getMaxTxid();
        String fsimageFilePath = "/Users/jinsimin/dfs/backupnode/" + filename + ".meta";
        lastFSImageFile = fsimageFilePath;
        fsimageFile = new RandomAccessFile(fsimageFilePath, "rw");
        fileOutputStream = new FileOutputStream(fsimageFile.getFD());
        channel = fileOutputStream.getChannel();
        channel.position(channel.size());
        channel.write(byteBuffer);
        channel.force(false); //强制把数据刷入磁盘
    }

    private void removeLastFsimageFile() {
        if (lastFSImageFile == null) {
            return;
        }
        File file = new File(lastFSImageFile);
        if (file.exists()) {
            file.delete();
        }
    }
}
